package com.xbai.spark.core.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * groupByKey 算子
  * 作用：groupByKey 也是对每个 key 进行操作，但只生成一个 sequence。
  * 按照 key 进行分组，直接进行 shuffle。
  * 案例：创建一个 pairRDD，将相同 key 对应值聚合到一个 sequence 中，并计算相同 key 对应值的相加结果。
 *
  * @author xbai
  * @Date 2020/12/30
  */
object Spark16_GroupByKey {

  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("groupByKey")
    val sc = new SparkContext(conf)

    val words = Array("one", "two", "two", "three", "three", "three")
    // 创建一个pairRDD
    val wordToOneRDD: RDD[(String, Int)] = sc.parallelize(words).map((_, 1))
    // 将相同key对应值聚合到一个sequence中
    val wordGroupRDD: RDD[(String, Iterable[Int])] = wordToOneRDD.groupByKey()
    // 计算相同key对应值的相加结果
    val wordToCountRDD: RDD[(String, Int)] = wordGroupRDD.map(t => (t._1, t._2.sum))
    wordToCountRDD.collect().foreach(println)
  }
}
